package top.wintp.offlinedataanalysis.anlyser.mr.nu;

import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

import top.wintp.offlinedataanalysis.anlyser.dim.StatsCommonDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.StatsUserDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.BrowserDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.DateDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.KpiDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.PlatformDimension;
import top.wintp.offlinedataanalysis.anlyser.value.map.TimeOutputValue;
import top.wintp.offlinedataanalysis.common.DateEnum;
import top.wintp.offlinedataanalysis.common.EventLogConstant;
import top.wintp.offlinedataanalysis.common.KpiType;

/**
 * description:
 * <p>
 *
 * @author: upuptop
 * <p>
 * qq: 337081267
 * <p>
 * CSDN:   http://blog.csdn.net/pyfysf
 * <p>
 * cnblogs:   http://www.cnblogs.com/upuptop
 * <p>
 * blog:   http://wintp.top
 * <p>
 * email:  pyfysf@163.com
 * <p>
 * time: 2019/08/2019/8/23
 * <p>
 */
public class NewInstallUserMapper extends TableMapper<StatsUserDimension, TimeOutputValue> {
    /**
     * HBase的列族
     */
    private byte[] family = Bytes.toBytes(EventLogConstant.EVENT_LOGS_FAMILY_NAME);

    /**
     * 浏览器新用户的kpi
     */
    private String browserKpi = KpiType.BROWSER_NEW_INSTALL_USER.name;
    /**
     * 新增用户的kpi
     */
    private String newInstallUserKpi = KpiType.NEW_INSTALL_USER.name;

    /**
     * out key
     */
    private StatsUserDimension statsUserDimension = new StatsUserDimension();

    /**
     * out value
     */
    private TimeOutputValue timeOutputValue = new TimeOutputValue();

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        //    读取HBase数据
        String serviceTime = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(family, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_SERVER_TIME))));
        String platform = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(family, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_PLATFORM))));
        String uuid = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(family, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_UUID))));
        String browserName = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(family, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_BROWSER_NAME))));
        String browserVersion = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(family, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_BROWSER_VERSION))));
        //    封装单一维度
        //    日期维度
        DateDimension dateDimension = DateDimension.buildDate(Long.valueOf(serviceTime), DateEnum.DAY);
        // 浏览器维度
        List<BrowserDimension> browserDimensions = BrowserDimension.buildList(browserName, browserVersion);
//        平台维度
        List<PlatformDimension> platformDimensions = PlatformDimension.buildList(platform);

//        为输出的 value 创建对象
        timeOutputValue.setId(uuid);
        timeOutputValue.setTime(Long.valueOf(serviceTime));

        //默认浏览器维度
        BrowserDimension defaultBrowserDimension = new BrowserDimension("", "");

        //        组合浏览器 新增用户维度
        for (PlatformDimension platformDimension : platformDimensions) {
            //        kpi维度
            KpiDimension kpiDimension = new KpiDimension(newInstallUserKpi);

            //        通用维度
            StatsCommonDimension statsCommonDimension = new StatsCommonDimension(dateDimension, platformDimension, kpiDimension);
            statsUserDimension.setStatsCommon(statsCommonDimension);
            statsUserDimension.setBrowser(defaultBrowserDimension);
            //                写出数据
            context.write(statsUserDimension, timeOutputValue);

            //    组合单一维度
            for (BrowserDimension browserDimension : browserDimensions) {
                kpiDimension.setKpiName(browserKpi);
                statsUserDimension.setBrowser(browserDimension);

                statsCommonDimension.setKpi(kpiDimension);
                statsUserDimension.setStatsCommon(statsCommonDimension);

                //  写出数据
                context.write(statsUserDimension, timeOutputValue);
            }

        }


    }
}
